package cn.doitedu.api;


import beans.OrderBean;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/27
 * @Desc: 学大数据，上多易教育
 * 有如下订单数据:
 * 订单id,商品id,品类id,金额,订单创建时间
 * o1,p1,category_01,1800,1709027697000
 * o1,p2,category_01,1600,1709027697000
 * o1,p3,category_02,1600,1709027697000
 * o2,p3,category_02,1600,1709027697000
 * o3,p4,category_03,1200,1709027697000
 * <p>
 * 统计:
 * 1. 最近 5分钟的成交总额，要求每 1分钟输出一次结果；
 * 2. 每 5分钟，各商品 品类的下单数;
 * 3. 每 5分钟，成交金额最高的前3种商品及其成交金额
 **/
public class _26_Exercise_Window_1 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        DataStreamSource<String> stream = env.socketTextStream("localhost", 8899);

        DataStream<OrderBean> mapped = stream.map(s -> {
            String[] split = s.split(",");
            return new OrderBean(split[0], split[1], split[2], Double.parseDouble(split[3]), Long.parseLong(split[4]));
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<OrderBean>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderBean>() {
                            @Override
                            public long extractTimestamp(OrderBean element, long recordTimestamp) {
                                return element.getCreate_time();
                            }
                        }));


        mapped.windowAll(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
                //.sum("money")
//                .reduce(new ReduceFunction<OrderBean>() {
//                    @Override
//                    public OrderBean reduce(OrderBean value1, OrderBean value2) throws Exception {
//                        double money1 = value1.getMoney();
//                        double money2 = value2.getMoney();
//
//                        return new OrderBean(value1.getOrderId(),value1.getProductId(),value1.getCategoryId(),money1+money2,value1.getCreate_time());
//                    }
//                })
//                .reduce(new ReduceFunction<OrderBean>() {
//                    @Override
//                    public OrderBean reduce(OrderBean b1, OrderBean b2) throws Exception {
//                        double m1 = b1.getMoney();
//                        double m2 = b2.getMoney();
//
//                        // minby算子
//                        /* if(m1 < m2){
//                            return value1;
//                        }else{
//                            return value2;
//                        }*/
//
//                        // min算子
//                        double min = m1 < m2 ? m1 : m2;
//                        return new OrderBean(b1.getOrderId(), b1.getProductId(), b1.getCategoryId(), min,b1.getCreate_time());
//                    }
//                })
                .aggregate(new AggregateFunction<OrderBean, AvgAcc, Double>() {
                    @Override
                    public AvgAcc createAccumulator() {
                        return new AvgAcc(0, 0);
                    }

                    @Override
                    public AvgAcc add(OrderBean value, AvgAcc accumulator) {

                        // 数量更新
                        accumulator.setCount(accumulator.getCount() + 1);

                        // 金额之和也更新
                        accumulator.setSum(accumulator.getSum() + value.getMoney());

                        return accumulator;
                    }

                    @Override
                    public Double getResult(AvgAcc accumulator) {

                        return accumulator.getSum()/ accumulator.getCount();
                    }

                    // 两个acc累加器的数据进行合并
                    @Override
                    public AvgAcc merge(AvgAcc a, AvgAcc b) {
                        int count = a.getCount() + b.getCount();
                        double sum = a.getSum() + b.getSum();

                        return new AvgAcc(count,sum);
                    }
                })


                .print();

        env.execute();


    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
class AvgAcc {
    private int count;
    private double sum;
}
